package com.smile.cloud.points.consume;

import com.smile.cloud.points.consume.core.DedupConfig;
import com.smile.cloud.points.service.PointsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;

/**
 * @author LGC
 */
@Slf4j
@Configuration
public class OrderConsumer {

    public static final String CONSUMER_GROUP = "order-consumer-group";
    public static final String TOPIC = "order";
    @Autowired
    private ConsumerConfig consumerConfig;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Bean
//    @ConditionalOnProperty(prefix = "rocketmq.consumer", value = "isOnOff", havingValue = "on")
    public DefaultMQPushConsumer defaultConsumer() {
        log.info(CONSUMER_GROUP + " consumer 正在创建---------------------------------------");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setNamesrvAddr(consumerConfig.getNameServerAddr());
        consumer.setConsumeThreadMin(consumerConfig.getConsumeThreadMin());
        consumer.setConsumeThreadMax(consumerConfig.getConsumeThreadMax());
        consumer.setConsumeMessageBatchMaxSize(consumerConfig.getConsumeMessageBatchMaxSize());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        try {
            consumer.subscribe(TOPIC, "*");
            consumer.registerMessageListener(orderListener());
            consumer.start();
            log.info(CONSUMER_GROUP + " consumer 创建成功 groupName={}, topics={}, nameServerAddr={}", CONSUMER_GROUP, TOPIC, consumerConfig.getNameServerAddr());
        } catch (Exception e) {
            log.error(CONSUMER_GROUP + " consumer 创建失败!{0}", e);
        }
        return consumer;
    }

    /**
     * redis 消息重复消费判断
     *
     * @return
     */
    @Bean
    public DedupConfig dedupConfig() {
        return DedupConfig.enableDedupConsumeConfig(CONSUMER_GROUP, stringRedisTemplate);
    }

    @Bean
    @ConditionalOnClass(value = {DedupConfig.class, PointsService.class})
    public OrderListener orderListener() {
        return new OrderListener(dedupConfig());
    }
}
